#include <iostream>
#include <future>
#include <thread>
#include <memory>
#include <atomic>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <vector>
#include <unistd.h>
#include <assert.h>

/*
    这是一个可以获取异步函数结果的线程池(生产者消费者模型)
    使用：
    1. future + package_task 获取异步函数结果
    2. mutex + condition_variable 来控制这个线程池的同步互斥关系
    3. std::vector<std::thread> _threads 一个管理线程池中任务线程的容器
    4. std::atomic<bool> _running 一个用于描述当前线程池是否运行的原子性成员
    5.
    using Func = std::function<void(void)>;
    std::vector<Func> _taskq; 一个用于存储任务的队列
*/

class threadpool
{
public:
    using Func = std::function<void(void)>;

    // 传入一个函数 + 不定参数
    // push内部会将传入的函数封装为一个异步任务(package_task)
    // 使用lambda包装(异步任务的智能指针解引用), 放入任务池中，等待工作线程运行
    template<typename F, typename ...Args>
    auto Push(const F &&func, Args&& ...args) -> std::future<decltype(func(args...))>
    {
        assert(_running != false);
        using rtyp = decltype(func(args...));
        auto tmp_func = std::bind(std::forward<F>(func), std::forward<Args>(args)...);
        auto task = std::make_shared<std::packaged_task<rtyp()>>(tmp_func);
        std::future<rtyp> fu = task->get_future();
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _taskq.push_back([task](){ (*task)(); });
            _cv.notify_one();
        }
        return fu;
    }

    void Stop()
    {
        if (_running == false)
            return;
        std::cout << "stop" << std::endl;
        _running = false;
        // 唤醒所有线程
        _cv.notify_all();
        // 进行回收
        for (auto &thread : _threads)
        {
            thread.join();
        }
    }

    // 获取单例实例（线程安全）
    static threadpool& GetInstance(const int threadnum = 1) 
    {
        static threadpool instance(threadnum); // C++11 保证初始化线程安全
        return instance;
    }

private:
    threadpool(const int threadnum = 1)
        : _running(true)
    {
        for (int i = 0; i < threadnum; ++i)
        {
            // 成员函数作为线程入口需要显式传递对象实例（this）
            _threads.emplace_back(&threadpool::Entry, this);
        }
    }

    threadpool(const threadpool&) = delete;
    threadpool& operator=(const threadpool&) = delete;

    ~threadpool()
    {
        Stop();
    }

    void Entry()
    {
        while (_running)
        {
            std::vector<Func> tmp_taskq;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _cv.wait(lock, [this]()
                         { return !_running || !_taskq.empty(); });
                // 取出任务执行, 要注意取出任务执行可以不用加锁
                tmp_taskq.swap(_taskq);
            }
            for (auto &task : tmp_taskq)
            {
                task();
            }
        }
    }

    std::atomic<bool> _running;
    std::mutex _mutex;
    std::condition_variable _cv;
    std::vector<std::thread> _threads; // 注意 _mutex 和 _cv 需要放在_threads之前，防止没有初始化出现线程安全问题
    std::vector<Func> _taskq;
};



int Function(int num, int change)
{
    std::cout << "异步函数" << " ";
    return num * change;
}


int main()
{
    threadpool& tp = threadpool::GetInstance();
    for(int i = 0; i < 10; i++)
    {
        std::future<int> res = tp.Push(Function, 10, i);
        std::cout << res.get() << std::endl;
    }

    tp.Stop();

    return 0;
}